package com.atguigu.gmall.realtime.dws.app;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import com.atguigu.gmall.realtime.common.base.BaseApp;
import com.atguigu.gmall.realtime.common.bean.TradeSkuOrderBean;
import com.atguigu.gmall.realtime.common.constant.Constant;
import com.atguigu.gmall.realtime.common.function.BeanToJsonMapFunction;
import com.atguigu.gmall.realtime.common.function.DimRichMapFunction;
import com.atguigu.gmall.realtime.common.util.DateFormatUtil;
import com.atguigu.gmall.realtime.common.util.FlinkSinkUtil;
import com.atguigu.gmall.realtime.common.util.HBaseUtil;
import com.atguigu.gmall.realtime.common.util.RedisUtil;
import io.lettuce.core.api.StatefulConnection;
import io.lettuce.core.api.StatefulRedisConnection;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.common.functions.ReduceFunction;
import org.apache.flink.api.common.functions.RuntimeContext;
import org.apache.flink.api.common.state.ValueState;
import org.apache.flink.api.common.state.ValueStateDescriptor;
import org.apache.flink.api.common.typeinfo.Types;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.datastream.AsyncDataStream;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.KeyedProcessFunction;
import org.apache.flink.streaming.api.functions.ProcessFunction;
import org.apache.flink.streaming.api.functions.async.ResultFuture;
import org.apache.flink.streaming.api.functions.async.RichAsyncFunction;
import org.apache.flink.streaming.api.functions.windowing.ProcessWindowFunction;
import org.apache.flink.streaming.api.windowing.assigners.TumblingEventTimeWindows;
import org.apache.flink.streaming.api.windowing.time.Time;
import org.apache.flink.streaming.api.windowing.windows.TimeWindow;
import org.apache.flink.table.utils.DateTimeUtils;
import org.apache.flink.util.Collector;
import org.apache.flink.util.TimeUtils;
import org.apache.hadoop.hbase.client.AsyncConnection;

import java.time.Duration;
import java.util.Collections;
import java.util.concurrent.TimeUnit;

/**
 * 交易域sku粒度下单各窗口汇总表
 *
 * V1版本:  采用最基本的方式来实现整个过程。
 */
public class DwsTradeSkuOrderWindow_v4 extends BaseApp {
    public static void main(String[] args) {
        new DwsTradeSkuOrderWindow_v4().start(10019, 4, "dws_trade_sku_order_window_v1", Constant.TOPIC_DWD_TRADE_ORDER_DETAIL);
    }

    @Override
    public void handle(StreamExecutionEnvironment env, DataStreamSource<String> stream) {
        // 转换结构  直接将json字符串转换成 TradeSkuOrderBean
        SingleOutputStreamOperator<TradeSkuOrderBean> transmitDs= stream.process(
                new ProcessFunction<String, TradeSkuOrderBean>() {
                    @Override
                    public void processElement(String value, ProcessFunction<String, TradeSkuOrderBean>.Context context, Collector<TradeSkuOrderBean> collector) throws Exception {
                        if (value != null) {
                            JSONObject jsonObj = JSON.parseObject(value);
                            // 转换结构
                            TradeSkuOrderBean bean = TradeSkuOrderBean.builder()
                                    .orderDetailId(jsonObj.getString("id"))
                                    .skuId(jsonObj.getString("sku_id"))
                                    .originalAmount(jsonObj.getBigDecimal("split_original_amount"))
                                    .orderAmount(jsonObj.getBigDecimal("split_total_amount"))
                                    .activityReduceAmount(jsonObj.getBigDecimal("split_activity_amount"))
                                    .couponReduceAmount(jsonObj.getBigDecimal("split_coupon_amount"))
                                    .ts(jsonObj.getLong("ts") * 1000)
                                    .build();
                            collector.collect(bean);
                        }
                    }
                }
        );
        // transmitDs.print("TRANSMIT");


        // 去重
        // 重复的现象:  上游DWD下单事务事实表的处理过程中，使用了 left join的操作，
        //            如果左表的数据先到， 会先写出一条未关联的数据(+I)
        //            等到右表的数据到到达， 会写出一条回撤的数据(-D) , 再写出一条关联成功的数据(+I)
        //            两次+I的数据就是重复的数据
        // 例如:       左表    detail_id   order_amount      右表  order_detail_id   xxx_amount
        //            +I      1001          5000.00                  null           null
        //            -D
        //            +I      1001          5000.00                  1001          2000.00

        // 去重方式:
        //   方式一:  可以通过会话窗口进行去重。  缺点时时效性太差， 同一个订单详情所有的数据到达后 ，还需要再等待gap的时间才可以向下游写出
        //           需要在数据中添加一个可以表示先后的标识（可以记录一下join时的时间） ，将同一个订单详情的数据收集到会话窗口中
        //           等到窗口计算时， 只需要将 标识 最大的数据(最全的数据写出即可)

        //  方式二 :  状态 + 定时器  ， 缺点时效性太差， 就算join后的数据几乎同时到达了， 也需要等待定时器触发
        //           需要在数据中添加一个可以表示先后的标识（可以记录一下join时的时间） , 通过状态维护到达的数据，
        //           第一条数据到达后，存入到状态中，并注册5秒后的定时器
        //           每次新到的数据要与状态中维护的数据进行比较，将大的保留在状态中。 等到定时器触发，将状态中的数据向下游写出

        // 方式三  :  状态 + 抵消
        //           第一条+I的数据到达后， 直接写出到下游， 并将该数据存入到状态中
        //           第二条+I的数据到达后， 将状态中的数据取出， 从第二条+I的数据中将度量值 减去 状态中存储数据中的度量值， 再将数据写出到下游、

        //          +I      1001    5000.00         null   null
        //          抵消     1001   -5000.00         null   null
        //          +I      1001    5000.00         1001  2000.00

        //     优化:
        //          +I         1001    5000.00                   null   null
        //          +I抵消      1001    5000.00 - 5000.00         1001  2000.00

        // 方式四  : 如果下游统计所需要的度量值都在左表中， 不需要进行去重， 直接写出第一条数据即可。
        SingleOutputStreamOperator<TradeSkuOrderBean> distinctDs = transmitDs.keyBy(TradeSkuOrderBean::getOrderDetailId)
                .process(
                        new KeyedProcessFunction<String, TradeSkuOrderBean, TradeSkuOrderBean>() {
                            private ValueState<TradeSkuOrderBean> lastTradeSkuOrderBeanState;

                            @Override
                            public void open(Configuration parameters) throws Exception {
                                RuntimeContext runtimeContext = getRuntimeContext();
                                lastTradeSkuOrderBeanState = runtimeContext.getState(
                                        new ValueStateDescriptor<TradeSkuOrderBean>("lastTradeSkuOrderBeanStateDesc", Types.POJO(TradeSkuOrderBean.class))
                                );
                            }

                            @Override
                            public void processElement(TradeSkuOrderBean tradeSkuOrderBean, KeyedProcessFunction<String, TradeSkuOrderBean, TradeSkuOrderBean>.Context context, Collector<TradeSkuOrderBean> collector) throws Exception {
                                TradeSkuOrderBean lastTradeSkuOrderBean = lastTradeSkuOrderBeanState.value();
                                lastTradeSkuOrderBeanState.update(tradeSkuOrderBean);
                                if (lastTradeSkuOrderBean == null) {
                                    // 第一条数据
                                    collector.collect(tradeSkuOrderBean);
                                } else {
                                    // 后续数据
                                    lastTradeSkuOrderBean.setOriginalAmount(lastTradeSkuOrderBean.getOriginalAmount().subtract(tradeSkuOrderBean.getOriginalAmount()));
                                    lastTradeSkuOrderBean.setOrderAmount(lastTradeSkuOrderBean.getOrderAmount().subtract(tradeSkuOrderBean.getOrderAmount()));
                                    lastTradeSkuOrderBean.setActivityReduceAmount(lastTradeSkuOrderBean.getActivityReduceAmount().subtract(tradeSkuOrderBean.getActivityReduceAmount()));
                                    lastTradeSkuOrderBean.setCouponReduceAmount(lastTradeSkuOrderBean.getCouponReduceAmount().subtract(tradeSkuOrderBean.getCouponReduceAmount()));
                                    collector.collect(lastTradeSkuOrderBean);
                                }
                            }
                        }
                );

        // distinctDs.print("DISTINCT");


        // 分配水位线
        SingleOutputStreamOperator<TradeSkuOrderBean> watermarksDs = distinctDs.assignTimestampsAndWatermarks(
                WatermarkStrategy.<TradeSkuOrderBean>forBoundedOutOfOrderness(Duration.ofSeconds(3L))
                        .withTimestampAssigner((bean, ts) -> bean.getTs())
                        .withIdleness(Duration.ofSeconds(120L))
        );


        // 按照 skuId 分组， 开窗 、 聚合
        SingleOutputStreamOperator<TradeSkuOrderBean> reduceDs = watermarksDs.keyBy(TradeSkuOrderBean::getSkuId)
                .window(TumblingEventTimeWindows.of(Time.seconds(5L)))
                .reduce(
                        new ReduceFunction<TradeSkuOrderBean>() {
                            @Override
                            public TradeSkuOrderBean reduce(TradeSkuOrderBean t1, TradeSkuOrderBean t2) throws Exception {
                                t1.setOriginalAmount(t1.getOriginalAmount().add(t2.getOriginalAmount()));
                                t1.setOrderAmount(t1.getOrderAmount().add(t2.getOrderAmount()));
                                t1.setActivityReduceAmount(t1.getActivityReduceAmount().add(t2.getActivityReduceAmount()));
                                t1.setCouponReduceAmount(t1.getCouponReduceAmount().add(t2.getCouponReduceAmount()));
                                return t1;
                            }
                        },
                        new ProcessWindowFunction<TradeSkuOrderBean, TradeSkuOrderBean, String, TimeWindow>() {
                            @Override
                            public void process(String key, ProcessWindowFunction<TradeSkuOrderBean, TradeSkuOrderBean, String, TimeWindow>.Context context, Iterable<TradeSkuOrderBean> iterable, Collector<TradeSkuOrderBean> collector) throws Exception {
                                TradeSkuOrderBean bean = iterable.iterator().next();
                                String start = DateFormatUtil.tsToDateTime(context.window().getStart());
                                String end = DateFormatUtil.tsToDateTime(context.window().getEnd());
                                String today = DateFormatUtil.tsToDate(context.window().getStart());
                                bean.setStt(start);
                                bean.setEdt(end);
                                bean.setCurDate(today);
                                collector.collect(bean);
                            }
                        }
                );

        // reduceDs.print("REDUCE");




        // 维度关联
        // 基于HBase的API从维度表中取出维度数据， 与当前流中的数据进行关联
        // dim_sku_info
        SingleOutputStreamOperator<TradeSkuOrderBean> addSkuInfoAsyncDs = AsyncDataStream.unorderedWait(
                reduceDs,
                new RichAsyncFunction<TradeSkuOrderBean, TradeSkuOrderBean>() {
                    private AsyncConnection hbaseAsyncConnection;
                    private StatefulRedisConnection<String, String> redisAsyncConnection;

                    @Override
                    public void open(Configuration parameters) throws Exception {
                        hbaseAsyncConnection = HBaseUtil.getAsyncConnection();
                        redisAsyncConnection = RedisUtil.getAsyncConnection();
                    }

                    @Override
                    public void close() throws Exception {
                        HBaseUtil.closeAsyncConnection(hbaseAsyncConnection);
                        RedisUtil.closeAsyncConnection(redisAsyncConnection);
                    }

                    @Override
                    public void asyncInvoke(TradeSkuOrderBean bean, ResultFuture<TradeSkuOrderBean> resultFuture) throws Exception {
                        String rowKey = bean.getSkuId();
                        String tableName = "dim_sku_info";
                        // 从redis中读取数据
                        JSONObject dimJsonObj = RedisUtil.readDimAsync(redisAsyncConnection, tableName, rowKey);
                        if (dimJsonObj == null) {
                            // 读不到去hbase读取维度数据
                            dimJsonObj = HBaseUtil.getAsyncRow(hbaseAsyncConnection, Constant.HBASE_NAMESPACE, tableName, rowKey, JSONObject.class, false);
                            if (dimJsonObj.size() > 0) {
                                // 写入到redis
                                RedisUtil.writeDimAsync(redisAsyncConnection, tableName, rowKey, dimJsonObj);
                            }
                        }
                        // 进行维度关联
                        if (dimJsonObj.size() > 0) {
                            bean.setSkuName(dimJsonObj.getString("sku_name"));
                            bean.setSpuId(dimJsonObj.getString("spu_id"));
                            bean.setTrademarkId(dimJsonObj.getString("tm_id"));
                            bean.setCategory3Id(dimJsonObj.getString("category3_id"));
                        }
                        // 将数据写出
                        resultFuture.complete(Collections.singleton(bean));
                    }
                },
                20,
                TimeUnit.SECONDS
        );
        addSkuInfoAsyncDs.print("addSkuInfoAsyncDs");



        /*SingleOutputStreamOperator<TradeSkuOrderBean> addSpuInfoAsyncDs = AsyncDataStream.unorderedWait(
                addSkuInfoAsyncDs,
                new RichAsyncFunction<TradeSkuOrderBean, TradeSkuOrderBean>() {
                    private AsyncConnection hbaseAsyncConnection;
                    private StatefulRedisConnection<String, String> redisAsyncConnection;

                    @Override
                    public void open(Configuration parameters) throws Exception {
                        hbaseAsyncConnection = HBaseUtil.getAsyncConnection();
                        redisAsyncConnection = RedisUtil.getAsyncConnection();
                    }

                    @Override
                    public void close() throws Exception {
                        HBaseUtil.closeAsyncConnection(hbaseAsyncConnection);
                        RedisUtil.closeAsyncConnection(redisAsyncConnection);
                    }

                    @Override
                    public void asyncInvoke(TradeSkuOrderBean bean, ResultFuture<TradeSkuOrderBean> resultFuture) throws Exception {
                        String rowKey = bean.getSkuId();
                        String tableName = "dim_spu_info";
                        // 从redis中读取数据
                        JSONObject dimJsonObj = RedisUtil.readDimAsync(redisAsyncConnection, tableName, rowKey);
                        if (dimJsonObj == null) {
                            // 读不到去hbase读取维度数据
                            dimJsonObj = HBaseUtil.getAsyncRow(hbaseAsyncConnection, Constant.HBASE_NAMESPACE, tableName, rowKey, JSONObject.class, false);
                            if (dimJsonObj.size() > 0) {
                                // 写入到redis
                                RedisUtil.writeDimAsync(redisAsyncConnection, tableName, rowKey, dimJsonObj);
                            }
                        }
                        // 进行维度关联
                        if (dimJsonObj.size() > 0) {
                            bean.setSpuName( dimJsonObj.getString("spu_name"));
                        }
                        // 将数据写出
                        resultFuture.complete(Collections.singleton(bean));
                    }
                },
                20,
                TimeUnit.SECONDS
        );


        addSpuInfoAsyncDs.print("addSpuInfoAsyncDs");*/



        /*// dim_base_trademark
        SingleOutputStreamOperator<TradeSkuOrderBean> addBaseTrademarkDs = addSpuInfoDs.map(
                new DimRichMapFunction<TradeSkuOrderBean>() {
                    @Override
                    public String getRowKey(TradeSkuOrderBean bean) {
                        return bean.getTrademarkId();
                    }

                    @Override
                    public void addDims(TradeSkuOrderBean bean, JSONObject dimJsonObj) {
                        bean.setTrademarkName(dimJsonObj.getString("tm_name"));
                    }

                    @Override
                    public String getTableName() {
                        return "dim_base_trademark";
                    }
                }
        );
        // addBaseTrademarkDs.print("addBaseTrademarkDs");
        // dim_base_category3
        SingleOutputStreamOperator<TradeSkuOrderBean> addBaseCategory3Ds = addBaseTrademarkDs.map(
                new DimRichMapFunction<TradeSkuOrderBean>() {
                    @Override
                    public String getRowKey(TradeSkuOrderBean bean) {
                        return bean.getCategory3Id();
                    }

                    @Override
                    public void addDims(TradeSkuOrderBean bean, JSONObject dimJsonObj) {
                        bean.setCategory3Name(dimJsonObj.getString("name"));
                        bean.setCategory2Id( dimJsonObj.getString("category2_id"));
                    }

                    @Override
                    public String getTableName() {
                        return "dim_base_category3";
                    }
                }
        );
        // addBaseCategory3Ds.print("addBaseCategory3Ds");


        // dim_base_category2
        SingleOutputStreamOperator<TradeSkuOrderBean> addBaseCategory2Ds = addBaseCategory3Ds.map(
                new DimRichMapFunction<TradeSkuOrderBean>() {
                    @Override
                    public String getRowKey(TradeSkuOrderBean bean) {
                        return bean.getCategory2Id();
                    }

                    @Override
                    public void addDims(TradeSkuOrderBean bean, JSONObject dimJsonObj) {
                        bean.setCategory2Name(dimJsonObj.getString("name"));
                        bean.setCategory1Id( dimJsonObj.getString("category1_id"));
                    }

                    @Override
                    public String getTableName() {
                        return "dim_base_category2";
                    }
                }
        );
        // addBaseCategory2Ds.print("addBaseCategory2Ds");
        // dim_base_category1
        SingleOutputStreamOperator<TradeSkuOrderBean> addBaseCategory1Ds = addBaseCategory2Ds.map(
                new DimRichMapFunction<TradeSkuOrderBean>() {
                    @Override
                    public String getRowKey(TradeSkuOrderBean bean) {
                        return bean.getCategory1Id();
                    }

                    @Override
                    public void addDims(TradeSkuOrderBean bean, JSONObject dimJsonObj) {
                        bean.setCategory1Name(dimJsonObj.getString("name"));
                    }

                    @Override
                    public String getTableName() {
                        return "dim_base_category1";
                    }
                }
        );

        addBaseCategory1Ds.print("DIM");

        // 写出到doris
        addBaseCategory1Ds.map(
                new BeanToJsonMapFunction<>()
        ).sinkTo(
                FlinkSinkUtil.getDorisSink( Constant.DORIS_DATABASE , Constant.DWS_TRADE_SKU_ORDER_WINDOW )
        );*/

    }
}